package main.java.com.lee.pattern.masterworker;

import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;

/**
 * @Description TODO
 * @Author winston
 * @DateTime 2022/2/18
 */
public class Worker<T extends Task, R> {
    private final LinkedBlockingQueue<T> taskQueue = new LinkedBlockingQueue<>();
    static AtomicInteger index = new AtomicInteger(1);
    private int workerId;
    private Thread thread;

    public Thread getThread() {
        return thread;
    }

    public Worker(Executor executor) {
//        thread = new Thread(this::run);
        executor.execute(this::run);
//        thread.start();
    }

    public void run(){
        if (thread == null) {
            thread = Thread.currentThread();
            System.out.println(thread.getName());
        }
        while (!Thread.currentThread().isInterrupted()) {
            try {
                T task = this.taskQueue.take();
                task.setWorkerId(workerId);
                task.execute();
            } catch (InterruptedException e) {
                System.out.println("worker " + workerId + " is interrupted");
            }
        }
    }

    public void submit(T task, Consumer<R> action) {
        task.resultAction = action;
        try {
            this.taskQueue.put(task);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
